package com.roy.kfk.basic;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @auth roykingw
 */

/**
 * 1. kafka中的拦截器有什么用？
 *      ⽣产者拦截机制 允许客户端 在⽣产者在消息发送到Kafka集群之前，对消息进⾏拦截，甚⾄可以修改消息内容
 * 2. 我们怎么使用 kafka的拦截机制呢？
 *      - 就是我们要自定义一个类，实现 ProducerInterceptor 接口
 *        重写相关方法就可以了
 *        如下：
 * 3. 那这个拦截器到底在业务中有什么用呢？
 *      - 在发送消息之前，可以添加一些新的数据，例如发送时间等
 */
public class MyInterceptor implements ProducerInterceptor {
    //1. 生产者 发送消息的到kafka之前 拦截
    @Override
    public ProducerRecord onSend(ProducerRecord producerRecord) {
        System.out.println("生产者 发送消息的到kafka之前 拦截消息 : " + producerRecord.toString());
        return producerRecord;
    }

    //2. 客户端 收到服务端 响应时触发
    // 这个响应可以用在 处理 消息重试
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        System.out.println("客户端 收到服务端 响应时触发：acknowledgement recordMetadata:"+recordMetadata.toString());
    }

    //3. 连接关闭时触发
    @Override
    public void close() {
        System.out.println("连接关闭时触发：producer closed");
    }




    //整理配置项
    @Override
    public void configure(Map<String, ?> map) {
        System.out.println("=====config start======");
        for (Map.Entry<String, ?> entry : map.entrySet()) {
            System.out.println("entry.key:"+entry.getKey()+" === entry.value: "+entry.getValue());
        }
        System.out.println("=====config end======");
    }
}
